package com.sgcc.dlsc.kafka.consumer.entity;

import com.sgcc.dlsc.kafka.consumer.util.ResultUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

/**
 * @Author: 李孟帅
 * @CreateTime: 2020-01-10 09:48
 * @Description: ${Description}
 */
public class ConsumeTask implements Runnable {
    private HttpHeaders headers = new HttpHeaders();
    private ConcurrentLinkedQueue<ConsumerRecords<String, String>> recordsQueue;
    private ConcurrentLinkedQueue<Map<TopicPartition, OffsetAndMetadata>> offsetQueue;
    private ConcurrentMap<TopicPartition, OffsetAndMetadata> offsetMap;
    private RestTemplate restTemplate;

    public ConsumeTask(ConcurrentLinkedQueue<ConsumerRecords<String, String>> recordsQueue, RestTemplate restTemplate,
                       ConcurrentLinkedQueue<Map<TopicPartition, OffsetAndMetadata>> offsetQueue,
                       ConcurrentMap<TopicPartition, OffsetAndMetadata> offsetMap) {
        this.recordsQueue = recordsQueue;
        this.restTemplate = restTemplate;
        this.offsetQueue=offsetQueue;
        this.offsetMap=offsetMap;
    }


    @Override
    public void run() {
        while (true) {
            if (recordsQueue.isEmpty()) {
                continue;
            }
            ConsumerRecords<String, String> records = recordsQueue.poll();
            Set<TopicPartition> partitions = records.partitions();
            for (TopicPartition partition : partitions) {
                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                Map<String, List<String>> data = new HashMap<>();
                String topic = partition.topic();
                List<String> messages = new ArrayList<>();
                for (ConsumerRecord<String, String> record : partitionRecords) {
                    String value = record.value();
                    messages.add(value);
                }
                data.put(topic, messages);
                headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
                HttpEntity<Map<String, List<String>>> requestEntity = new HttpEntity<>(data, headers);
                try {
                    ResponseEntity<ResultUtil> responseEntity = restTemplate.exchange("http://localhost:16000/syncListLog", HttpMethod.POST, requestEntity, ResultUtil.class);
                    ResultUtil body = responseEntity.getBody();
                    ConsumerRecord<String, String> stringStringConsumerRecord = partitionRecords.get(partitionRecords.size() - 1);
                    long offset = stringStringConsumerRecord.offset();
                    offsetMap.put(partition, new OffsetAndMetadata(offset + 1));
                    offsetQueue.add(offsetMap);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
